package com.atguigu.kafka.produce.demo;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Arrays;
import java.util.Properties;

/**
 * @ClassName MyProduce
 * @DeScription MyProduce
 * @Author 剧情再美终是戏
 * @Date 2019/12/4 10:45
 * @Version 1.0
 **/
public class MyProduce {

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("acks", "all");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1000);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 自定义分区
//        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.atguigu.kafka.produce.demo.MyPartitioner");
        // 自定义拦截器
//        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList("com.atguigu.kafka.produce.demo.MyCountInterceptor", "com.atguigu.kafka.produce.demo.MyTimestampInterceptor"));

        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(props);
        for (int i = 0; i < 1000; i++) {
            ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("topic_start", null, i, "1ccccccaa" + i);
            producer.send(record);
            Thread.sleep(10);
            System.out.print(".");
        }

        producer.close();
    }
}
